package com.jzo2o.foundations.handler;

import com.jzo2o.canal.listeners.AbstractCanalRabbitMqMsgListener;
import com.jzo2o.es.core.ElasticSearchTemplate;
import com.jzo2o.foundations.constants.IndexConstants;
import com.jzo2o.foundations.model.domain.ServeSync;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;

/**
 * 接受从Canal发送的MQ消息并解析处理完成ES的索引更新
 * @author JIAN
 */
@Slf4j
@Component
public class ServeCanalDataSyncHandler extends AbstractCanalRabbitMqMsgListener<ServeSync> {
    /**
     * 监听MQ的消息
     */
    @RabbitListener(bindings = @QueueBinding(
            exchange = @Exchange(name = "exchange.canal-jzo2o", type = ExchangeTypes.TOPIC),
            value = @Queue(name = "canal-mq-jzo2o-foundations",
                    // 限制单消费者保证顺序性
                    arguments = @Argument(name = "x-single-active-consumer", value = "true", type = "java.lang.Boolean")),
            key = "canal-mq-jzo2o-foundations"),
            // 限制单进程消费保证顺序性
            concurrency = "1")
    public void onMessage(Message message) throws Exception {
        parseMsg(message);
    }

    @Resource
    private ElasticSearchTemplate elasticSearchTemplate;

    @Override
    public void batchSave(List<ServeSync> data) {
        Boolean isSuccess = elasticSearchTemplate.opsForDoc().batchInsert(IndexConstants.SERVE, data);
        log.warn("[batchSave] status: {} data: {}", isSuccess, data.toString());
        if (!isSuccess) {
            // 通过抛出异常自动发送unack消息
            throw new RuntimeException("同步失败");
        }
    }

    @Override
    public void batchDelete(List<Long> ids) {
        Boolean isSuccess = elasticSearchTemplate.opsForDoc().batchDelete(IndexConstants.SERVE, ids);
        log.warn("[batchDelete] status: {} ids: {}", isSuccess, ids.toString());
        if (!isSuccess) {
            // 通过抛出异常自动发送unack消息
            throw new RuntimeException("同步失败");
        }
    }
}